"""
author: liuxu
date: 2024.12.3
description:
"""
import re
import allure
import pytest

from common.logger_ctrl import mylogger
from common.ssh_command import exec_cmd
from base_public.testbase import BasePublic

@pytest.mark.high
@allure.feature('SQTT功能测试')
class TestSQTT(BasePublic):
    """测试sqtt功能"""
    def test_sqtt(self, test_env):
        """
        获取sqtt提供对核函数汇编代码的性能分析
        :return:
        """
        # 编译待测用例
        exe_path = self.compile_source_code(src_file_name="hipSimple/matrix.cpp")

        # 执行测试命令
        ret, output = exec_cmd(f"hipprof --sqtt {exe_path}", cwd=test_env)
        assert ret is True and self.check_sqtt_is_success(output, cwd=test_env) is True

    @pytest.mark.parametrize("sqtt_type", (0, 1, "all"))  # 参数all等价于2
    def test_sqtt_type(self, test_env, sqtt_type):  # TODO涉及到event时候会抛出std::map::at异常，待复现
        """
        --sqtt-type <type>
            输出指定类型的数据。
            stat 输出指令统计文件
            raw 输出原始 tokens 数据
            issue 输出指令发射数据
            event 输出时间数据
            wave 输出当前CU的 wave 信息
            all_wave 输出所有 CU上 wave信息
            stat_stall 输出 stall 信息
            stat_valu 输出 valu 信息

            其他特殊值：
            默认值 stat,wave,issue
            0 等价于 stat,wave,issue
            1 等价于 stat,wave,issue,stat_stall,stat_valu
            all 等价于 stat,issue,event,wave,all_wave,stat_stall,stat_valu
        :return:
        """
        # 编译待测用例
        exe_path = self.compile_source_code(src_file_name="hipSimple/matrix.cpp")

        # 执行测试命令
        ret, output = exec_cmd(f"hipprof --sqtt --sqtt-type {sqtt_type} {exe_path}", cwd=test_env)
        assert ret is True
        assert self.check_sqtt_is_success(output, cwd=test_env) is True

    @pytest.mark.parametrize("flush_mode", (0, 1))
    def test_sqtt_flush_mode(self, test_env, flush_mode):
        """
        --flush-mode <mode code>
        Submit commands to HW on every operation. 0 - Disable, 1 - Enable(default).
        每一个操作都直接提交到硬件。0：关闭，1：开启（默认值）
        :return:
        """
        # 编译待测用例
        exe_path = self.compile_source_code(src_file_name="hipSimple/matrix.cpp")

        # 执行测试命令
        ret, output = exec_cmd(f"hipprof --sqtt --flush-mode {flush_mode} {exe_path}", cwd=test_env)
        assert ret is True and self.check_sqtt_is_success(output, cwd=test_env) is True

    def test_sqtt_output_type_html(self, test_env):
        """
        --output-type <type>	0（默认）：导出结果为json文件，1：导出结果为html文件
        :return:
        """
        # 编译待测用例
        exe_path = self.compile_source_code(src_file_name="hipSimple/matrix.cpp")

        # 执行测试命令
        ret, output = exec_cmd(f"hipprof --sqtt --output-type 1 {exe_path}", cwd=test_env)
        assert ret is True and self.check_sqtt_is_success(output, cwd=test_env) is True

    def test_sqtt_filter_kernel_name(self, test_env):
        """
        测试仅抓取指定核函数的数据
        :return:
        """
        # 编译待测用例
        exe_path = self.compile_source_code(src_file_name="pmc_sqtt/pmc_optimization_test.cpp")

        # 执行测试命令
        kernel_name = "baseline_kernel"  # baseline_kernel,optimized_kernel
        ret, output = exec_cmd(f"hipprof --sqtt --kernel-name {kernel_name} {exe_path}", cwd=test_env)
        assert ret is True and self.check_sqtt_is_success(output, cwd=test_env) is True

        # 校验打印结果核函数只有baseline_kernel
        assert "baseline_kernel" in output and "optimized_kernel" not in output

    def test_sqtt_pmc_off(self, test_env):
        """
        测试启动时关闭pmc和sqtt抓取,通过添加 hipProfilerStart/Stop 调用在运行时开启或关闭 PMC/SQTT 抓取

        :return:
        """
        # 编译待测用例
        exe_path = self.compile_source_code(src_file_name="pmc_sqtt/pmc_optimization_test.cpp")

        # 执行测试命令
        ret, output = exec_cmd(f"hipprof --sqtt --pmc-off {exe_path}", cwd=test_env)
        assert ret is True and self.check_sqtt_is_success(output, cwd=test_env) is True

        # 校验打印结果核函数只有使用了hipProfilerStart开启后的optimized_kernel
        assert "baseline_kernel" not in output and "optimized_kernel" in output

    def test_sqtt_i_filter_indicator(self, test_env):
        """
        测试sqtt使用-i参数过滤指标
        :return:
        """
        cfg_name = "filter_config.txt"
        filter_info ="""pmc:GRBM_COUNT GRBM_GUI_ACTIVE SQ_INSTS_FLAT SQ_INSTS_SALU SQ_LDS_BANK_CONFLICT SQ_WAVES
        range: 0 1
        gpu: 0 1 2 3 4 5 6 7 
        kernel: advancedKernel simple  
        sqtt:MAX_SE=1 TARGET_CU=0 SIMD=1,3 MAX_WAVE=1
        sqttbufsize:5368709120"""
        case_path = self.compile_source_code(src_file_name="pmc_sqtt/pmc_sqtt_test.cpp", compile_paras="-O0 -g")
        with open(cfg_name, 'w', encoding='utf-8') as f:
            for line in filter_info.splitlines():
                f.write(line.strip()+"\n")
        exec_cmd(f"cat {cfg_name}")
        ret, output = exec_cmd(f"hipprof --sqtt --sqtt-type all -i {cfg_name} {case_path}", cwd=test_env)
        assert ret is True
        assert "HIP_PROF:finish" in output and re.search("fail|error|can not", output, re.DOTALL) is None
        assert self.exist_file("thread_trace_*.json") and self.exist_file("thread_trace_*.stat.html")
        assert ret is True and self.check_sqtt_is_success(output, check_gen_files=False, cwd=test_env) is True
        # TODO 打开json文件检查过滤指标是否生效

    @pytest.mark.medium
    def test_sqtt_empty_kernel(self, test_env):
        """
        测试分析空kernel的sqtt数据
        :param test_env:
        :return:
        """
        # 编译待测用例
        exe_path = self.compile_source_code(src_file_name="rtc/rtc_perf_test.cpp")

        # 执行测试命令
        expected_execution_time = 5  # 秒
        ret, output = exec_cmd(f"hipprof --hip-trace --sqtt {exe_path} 0 1", cwd=test_env, timeout = expected_execution_time)
        assert ret is True
        assert self.check_sqtt_is_success(output, cwd=test_env) is True

